package com.zhang.hadoop.flink.test6;

import com.zhang.hadoop.flink.base.ClickSource;
import com.zhang.hadoop.flink.base.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author: zhang yufei
 * @createTime:2022/7/9 17:06
 * @description:
 */
public class ProcessFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );

        stream.process(new ProcessFunction<Event, Object>() {
            @Override
            public void processElement(Event event, Context context, Collector<Object> collector) throws Exception {
                if(event.user.equals("欧阳慧")){
                    collector.collect(event.user+" clicks "+event.url);
                }else if (event.user.equals("陈玉萍")){
                    collector.collect(event.user);
                    collector.collect(event.user);
                }
                System.out.println("timestamp:"+context.timestamp());
                System.out.println("watermark:"+context.timerService().currentWatermark());

                System.out.println(getRuntimeContext().getIndexOfThisSubtask());
            }
        }).print();

        env.execute();
    }
}
